package top.cyuw.simplerpc.remoting.server;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import top.cyuw.simplerpc.constant.RpcConstants;
import top.cyuw.simplerpc.dto.RpcMessage;
import top.cyuw.simplerpc.dto.RpcRequest;
import top.cyuw.simplerpc.dto.RpcResponse;

/**
 * @author chen
 * @date 2023/3/13 12:28
 */
@Slf4j
public class SimpleRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final InvokeHandler invokeHandler;

    public SimpleRpcServerHandler(InvokeHandler invokeHandler) {
        this.invokeHandler = invokeHandler;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            if (msg instanceof RpcMessage) {
                RpcMessage rpcMessage = (RpcMessage) msg;
                byte messageType = rpcMessage.getMessageType();

                RpcMessage responseRpcMessage = new RpcMessage();
                if (messageType == RpcConstants.MESSAGE_TYPE_HEARTBEAT_PING) {
                    responseRpcMessage.setMessageType(RpcConstants.MESSAGE_TYPE_HEARTBEAT_PONG);
                } else if (messageType == RpcConstants.MESSAGE_TYPE_RPC_REQUEST) {
                    responseRpcMessage.setMessageType(RpcConstants.MESSAGE_TYPE_RPC_RESPONSE);
                    RpcRequest rpcRequest = (RpcRequest) rpcMessage.getBody();
                    RpcResponse rpcResponse = new RpcResponse();
                    rpcResponse.setRequestId(rpcRequest.getRequestId());
                    responseRpcMessage.setBody(rpcResponse);

                    Object result = invokeHandler.handle(rpcRequest);
                    rpcResponse.setData(result);
                }

                ctx.writeAndFlush(responseRpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("server exception: " + cause.getMessage(), cause);
        ctx.close();
    }

}
